package cn.com.greatwall.kafka.rest;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;

import javax.servlet.http.HttpServletResponse;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.data.domain.Pageable;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import cn.com.greatwall.common.exception.BadRequestException;
import cn.com.greatwall.kafka.domain.Cluster;
import cn.com.greatwall.kafka.domain.Topic;
import cn.com.greatwall.kafka.service.KafkaService;
import cn.com.greatwall.kafka.service.MonitorService;
import cn.com.greatwall.kafka.service.TopicService;
import cn.com.greatwall.kafka.service.dto.ClusterDto;
import cn.com.greatwall.kafka.service.dto.ConsumerDto;
import cn.com.greatwall.kafka.service.dto.KafkaRecordDto;
import cn.com.greatwall.kafka.service.dto.criteria.ClusterQueryCriteria;
import cn.com.greatwall.kafka.service.dto.criteria.TopicQueryCriteria;
import cn.com.greatwall.logging.annotation.Log;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.RequiredArgsConstructor;

/**
 * @Author herw
 * @Time 2021-01-21 13:23:33
 * @Version 1.0
 * @Description: TODO(用一句话描述该文件做什么)
 */
@RestController
@RequiredArgsConstructor
@Api(tags = "MQ：Kafka分类管理")
@RequestMapping("/api/mq/topic")
public class TopicController {
    private final TopicService topicService;
    private final KafkaService kafkaService;
    private final MonitorService monitorService;
    private static final String ENTITY_NAME = "task";

    @Log("导出分类")
    @ApiOperation("导出分类")
    @GetMapping(value = "/download")
    @PreAuthorize("@el.check('topic:list')")
    public void download(HttpServletResponse response, TopicQueryCriteria criteria) throws IOException {
        topicService.download(topicService.queryAll(criteria), response);
    }

    @Log("查询分类")
    @ApiOperation("查询分类")
    @GetMapping
    @PreAuthorize("@el.check('topic:list')")
    public ResponseEntity<Object> query(TopicQueryCriteria criteria, Pageable pageable) {
        return new ResponseEntity<>(topicService.queryAll(criteria, pageable), HttpStatus.OK);
    }
    
    @Log("查询分类")
    @ApiOperation("查询分类")
    @GetMapping(value = "/{id}")
    @PreAuthorize("@el.check('topic:list')")
    public ResponseEntity<Object> query(@PathVariable String id){
        return new ResponseEntity<>(topicService.findById(Long.parseLong(id)), HttpStatus.OK);
    }
    
    @Log("查询分类")
    @ApiOperation("查询集群的分类")
    @GetMapping(value = "/cluster/{clusterId}")
    @PreAuthorize("@el.check('cluster:list','cluster:add','cluster:edit')")
    public ResponseEntity<Object> getTopicList(@PathVariable String clusterId) throws InterruptedException, ExecutionException {
        return new ResponseEntity<>(monitorService.getTopicList(clusterId), HttpStatus.OK);
    }

    @Log("新增分类")
    @ApiOperation("新增分类")
    @PostMapping
    @PreAuthorize("@el.check('topic:add')")
    public ResponseEntity<Object> create(@Validated @RequestBody Topic resources) {
        if (resources.getId() != null) {
            throw new BadRequestException("A new " + ENTITY_NAME + " cannot already have an ID");
        }
        topicService.create(resources);
        return new ResponseEntity<>(HttpStatus.CREATED);
    }

    @Log("修改分类")
    @ApiOperation("修改分类")
    @PutMapping
    @PreAuthorize("@el.check('topic:edit')")
    public ResponseEntity<Object> update(@Validated @RequestBody Topic resources) {
        topicService.update(resources);
        return new ResponseEntity<>(HttpStatus.NO_CONTENT);
    }

    @Log("删除分类")
    @ApiOperation("删除分类")
    @DeleteMapping
    @PreAuthorize("@el.check('topic:del')")
    public ResponseEntity<Object> delete(@RequestBody Set<Long> ids) {
        topicService.delete(ids);
        return new ResponseEntity<>(HttpStatus.OK);
    }

    @Log("生产消息")
    @ApiOperation("生产消息")
    @PostMapping("/producer")
    @PreAuthorize("@el.check('topic:add')")
    public ResponseEntity<Object> producer(@RequestParam String clusterId, @RequestParam String topicName,
            @RequestParam String key, @RequestParam String value) {
        try {
            kafkaService.send(clusterId, topicName, key, value);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return new ResponseEntity<>(HttpStatus.OK);
    }
    
    @Log("生产消息")
    @ApiOperation("生产消息")
    @PostMapping("/consumer")
    @PreAuthorize("@el.check('topic:add')")
    public ResponseEntity<Object> consumer(@Validated @RequestBody ConsumerDto resources) {
        ConsumerRecords<String, String> records = kafkaService.consumer(resources);
        List<Object> retList = new ArrayList<>();
        for (ConsumerRecord<String, String> record: records) {
            KafkaRecordDto dto = new KafkaRecordDto();
            dto.setPartition(record.partition());
            dto.setOffset(record.offset());
            dto.setKey(record.key());
            dto.setValue(record.value());
            
            retList.add(dto);
        }
        return new ResponseEntity<>(retList, HttpStatus.OK);
    }
}
